package com.zhang.flink.example;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * @title: 使用reduce计算wordCount
 * @author: zhang
 * @date: 2022/2/11 19:07
 */
public class WordCount {
    public static void main(String[] args) throws Exception {
        //todo 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //env.setParallelism(1);

        // todo 从端口读取数据
        env
                .socketTextStream("hadoop102", 9999)
                // flatMap的语义：将列表或者流中的每一个元素，转换成0个，1个或者多个元素
                // flatMap是无状态的算子
                .flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
                    @Override
                    public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
                        String[] fields = value.split(" ");
                        for (String field : fields) {
                            out.collect(Tuple2.of(field, 1));
                        }
                    }
                })
                /**
                 * 使用key将数据shuffle到不同的逻辑分区；相同key的数据一定在同一个任务插槽（物理分区）中
                 * 不同key的数据也可能在同一个任务插槽（物理分区）中；
                 * 同一个逻辑分区的数据一定在同一个物理分区中
                 */
                .keyBy(r -> r.f0)
                /**
                 * reduce 内部维护累加器（但是reduce并没有暴露累计器定义规则），累加器的类型和输入的数据类型一致
                 * 当第一条元素到来的时候直接作为累加器输出
                 * 当下来数据来的时候和累加器做聚合（聚合逻辑是自己定义），聚合完之后输出累加器，更新完累加器之后，输入事件就被丢弃了--滚动聚合
                 * reduce函数定义的是输入元素和累加器的累加规则，累加完以后更新累加器
                 */
                .reduce(new ReduceFunction<Tuple2<String, Integer>>() {
                    @Override
                    public Tuple2<String, Integer> reduce(Tuple2<String, Integer> value1, Tuple2<String, Integer> value2) throws Exception {
                        return Tuple2.of(value1.f0, value1.f1 + value2.f1);
                    }
                })
                .print();


        // todo 执行任务
        env.execute();
    }
}
